package com.shujia.mr.wc1;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapTask;
import org.apache.hadoop.mapred.ReduceTask;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


/**
 * 大数据中计算组件的入门程序：单词统计
 * <p>
 * words.txt
 * 内容：
 * hello world java
 * hadoop hello shangping
 * dingyijie hello hadoop shangping dingyijie
 * java
 * <p>
 * 需求：使用MR统计文件中每个单词出现的次数
 * hello 3
 * world 1
 * java 2
 * ..
 * <p>
 * 因为Map和Reduce是两种不同的逻辑和阶段，所以我们要分别编写Map阶段的逻辑，以及Reduce阶段的逻辑
 * <p>
 * 将来一个mapreduce程序是要进行打包  xxxx.jar
 * 将jar包放入到hadooplinux集群中使用命令进行计算
 * <p>
 * hadoop jar xxx.jar MR的主类名 输入文件的路径 输出文件的路径
 * <p>
 * 编写一个MR程序的步骤：
 * 1、确保数据在HDFS上（或者运行之前产生）
 * 2、编写自定义Map类，目的是为了编写map任务要执行的逻辑，继承Mapper类，重写map方法
 * 3、编写自定义Reduce类，目的是为了编写reduce任务要执行的逻辑，继承Reducer类，重写reduce方法 （MR程序可以不需要Reduce,比如简单的数据清洗）
 * 4、编写主类，13行必须要写的代码
 */

public class WordCountDemo {
    public static void main(String[] args) throws Exception {
        //创建hadoop配置文件对象
        Configuration conf = new Configuration();
        conf.set("mapreduce.input.fileinputformat.split.minsize", "256L"); // 调整minSize的大小，最终会影响到一个切片的大小
        conf.set("mapreduce.task.io.sort.mb", "200"); //设置环形缓冲区大小
        conf.set("mapreduce.map.sort.spill.percent", "0.9F"); // 设置环形缓冲区溢写的比例

        //创建Job作业对象
        Job job = Job.getInstance(conf);
        job.setJobName("29期WordCount入门MR程序");
        //reduce的个数是可以进行调整的
//        job.setNumReduceTasks(4); // 如果不设置，整个作业默认只有一个reduce

        //设置主类名
        job.setJarByClass(WordCountDemo.class);

        //开启预聚合，一般情况下，预聚合的逻辑与reduce逻辑一致
        job.setCombinerClass(MyReducer.class);

        //设置Map类
        job.setMapperClass(MyMapper.class);
        //设置Reduce类
        job.setReducerClass(MyReducer.class);

        //设置Map任务输出的键和值的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置Reduce任务输出的键和值的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //设置文件输入的路径
        //hadoop jar xxx.jar MR的主类名 输入文件的路径 输出文件的路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        //设置文件输出的路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));


        //提交作业到yarn上面执行，由yarn进行分配资源
        boolean flag = job.waitForCompletion(true);
        //flag标志意思是执行的成功与否
        if (flag) {
            System.out.println("==================29期WordCount入门MR程序执行成功！！==================");
        } else {
            System.out.println("=========程序执行失败！！=====================");
        }

//        MapTask
//        ReduceTask

    }
}
